package org.nbict.iot.trend;

import org.apache.storm.generated.StormTopology;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.tuple.Fields;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.task.realtime.func.PacketJsonProject;
import org.nbict.iot.task.realtime.func.PacketResolver;
import org.nbict.iot.trident.kafka.KafkaSpout4ParsedTools;

/**
 * 报文趋势分析
 * Created by songseven on 18/6/29.
 */
public class LogAnalysisTrident {

    public static StormTopology buildTopology() {

        String url = PropertyUtil.getProperty("mongo.url");
        String database = PropertyUtil.getProperty("mongo.database");
        String username = PropertyUtil.getProperty("mongo.username");
        String password = PropertyUtil.getProperty("mongo.password");
        String admin = PropertyUtil.getProperty("mongo.admin.name");
        String collectionName = "vehicle_map";

        TridentTopology topology = new TridentTopology();

        KafkaSpoutConfig spoutConfig = KafkaSpout4ParsedTools.newKafkaSpoutConfig
                ("192.168.1.83:9092", "transit-.*-simple", // bootstrap server, 192.168.1.83:9092
                        "iot-car-group-gis-update"); //group id
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4ParsedTools
                .newKafkaTridentSpoutOpaque(spoutConfig);

        Stream stream = topology.newStream("iot-car-parsed", kafkaSpout);
        Fields packetFields = new Fields("car_packet_header",
                "car_packet_body");
        Fields outputFields = new Fields("car_vin", "action", "timestamp",
                "upload");

        Stream stage = stream.each(new Fields("platform_id", "packet"), new PacketJsonProject(packetFields), packetFields)
                .each(new Fields("platform_id", "car_packet_header", "car_packet_body"), new PacketResolver(), outputFields)
                .project(new Fields("platform_id", "car_vin", "action", "timestamp", "upload")); //去除packet元组

        EWMA ewma = new EWMA().sliding(1.0, EWMA.Time.MINUTES).withAlpha(EWMA.ONE_MINUTE_ALPHA);

        Stream averageStream = stage.each(new Fields("timestamp"), new MovingAverageFunction(ewma, EWMA.Time.MINUTES), new Fields("average"));

        return null;
    }


}
